package com.storm.lesson.trident;

import java.util.Random;

import storm.trident.TridentState;
import storm.trident.TridentTopology;
import storm.trident.operation.builtin.Count;
import storm.trident.operation.builtin.FilterNull;
import storm.trident.operation.builtin.FirstN;
import storm.trident.operation.builtin.MapGet;
import storm.trident.testing.FixedBatchSpout;
import storm.trident.testing.MemoryMapState;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.LocalDRPC;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.StormTopology;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

import com.storm.lesson.trident.functions.MySplit;
import com.storm.lesson.trident.functions.Split;

public class TridentPVTopo {

	public static StormTopology buildTopology(LocalDRPC drpc) {
		Random random = new Random();
		String[] hosts = { "www.taobao.com" };
		String[] session_id = { "ABYH6Y4V4SCVXTG6DPB4VH9U123",
				"XXYH6YCGFJYERTT834R52FDXV9U34",
				"BBYH61456FGHHJ7JL89RG5VV9UYU7", "CYYH6Y2345GHI899OFG4V9U567",
				"VVVYH6Y4V4SFXZ56JIPDPB4V678" };
		String[] time = { "2014-01-07 08:40:50", "2014-01-07 08:40:51",
				"2014-01-09 08:40:52", "2014-01-09 08:40:53",
				"2014-01-07 09:40:49", "2014-01-08 10:40:49",
				"2014-01-08 11:40:49", "2014-01-09 12:40:49" };

		/**
		 * spout是一个FixedBatchSpout对象，以eachLog作为输出字段名，最大的Batch大小为3，实现了从输入流中循环读取句子的功能。
		 */
		FixedBatchSpout spout = new FixedBatchSpout(new Fields("eachLog"), 3,// 最大的Batch大小为3
				new Values(hosts[0] + "\t" + session_id[random.nextInt(5)]// 每一个Values对应一个Tuple
						+ "\t" + time[random.nextInt(8)]), new Values(hosts[0]
						+ "\t" + session_id[random.nextInt(5)] + "\t"
						+ time[random.nextInt(8)]), new Values(hosts[0] + "\t"
						+ session_id[random.nextInt(5)] + "\t"
						+ time[random.nextInt(8)]), new Values(hosts[0] + "\t"
						+ session_id[random.nextInt(5)] + "\t"
						+ time[random.nextInt(8)]), new Values(hosts[0] + "\t"
						+ session_id[random.nextInt(5)] + "\t"
						+ time[random.nextInt(8)]), new Values(hosts[0] + "\t"
						+ session_id[random.nextInt(5)] + "\t"
						+ time[random.nextInt(8)]), new Values(hosts[0] + "\t"
						+ session_id[random.nextInt(5)] + "\t"
						+ time[random.nextInt(8)]), new Values(hosts[0] + "\t"
						+ session_id[random.nextInt(5)] + "\t"
						+ time[random.nextInt(8)]));
		// Spout.setCycle(true) 让这些数据循环产生，而不是产生一次就没了
		spout.setCycle(true);

		/**
		 * 首先，创建了一个TridentTopology对象,该对象公开了很多构建Trident计算的接口方法.
		 * TridentToplogy有一个newStream方法,newStream方法在拓扑中创建了一个新的数据流以便从输入源中读取数据.
		 * 本例中,输入源是之前定义的FixedBatchSpout对象。输入源也可以是Kestrel或者Kafka等消息队列代理。
		 * Trident在Zookeeper中为每个输入源保存保存少量的状态信息以跟踪数据处理情况，这里的spout1字符串指的是在Zookeeper中为Trident保存元数据的ZNode存储节点的名称。
		 * each方法的第一个输入参数是输入字段inputFields对象，第二个输入参数是对每个输入字段应用实现了Function接口的函数function,第三个输入参数是function的输出字段。each方法的返回类型是Stream对象。
		 * groupBy，数据流以date字段进行分组，然后每个组使用Count聚合器进行持久化聚合。
		 * persistentAggregate函数知道如何在一个状态源进行存储并更新聚合的结果。在本例中，将session_id的统计结果保存在内存中。
		 */
		TridentTopology topology = new TridentTopology();
		TridentState wordCounts = topology
				.newStream("spout1", spout)
				// .parallelismHint(16)
				.each(new Fields("eachLog"), new MySplit("\t"),
						new Fields("date", "session_id"))
				.groupBy(new Fields("date"))
				.persistentAggregate(new MemoryMapState.Factory(),
						new Fields("session_id"), new Count(), new Fields("PV"));
		// .parallelismHint(16);

		/**
		 * 使用相同的TridentTopology对象来创建DRPC流，并以GetPV作为函数名。该函数名对应于DRPCClient对象的execute方法的第一个参数的函数名。
		 * 每个DRPC请求被当成只包含一个元组作为输入的Batch来处理。元组只包含一个由客户端提供参数的args字段。在该例子中，该参数是一个由空格分隔的date列表
		 * 首先，使用Split函数把请求的参数分割成它的组成部分的date，并以date字段作为输出字段，发射元组。该流是以date字段进行分组
		 * stateQuery操作用于查询拓扑第一部分生成的TridentState对象，即上面的wordCounts。stateQuery以一个状态来源和一个用于查询状态的函数作为输入，由拓扑的其他部分来计算统计。
		 * MapGet函数获取每个date的计数
		 * 由于DRPC流使用与TridentState对象完全一样的分组方式(按照date字段进行分组)，每个date的查询结果会被路由到TridentState对象管理和更新这个date的分区去执行。
		 * 用FillterNull过滤器把那些没有统计到的date过滤出去
		 */
		topology.newDRPCStream("GetPV", drpc)
				.each(new Fields("args"), new Split(" "), new Fields("date"))
				.groupBy(new Fields("date"))
				.stateQuery(wordCounts, new Fields("date"), new MapGet(),
						new Fields("PV"))
				.each(new Fields("PV"), new FilterNull())
				.applyAssembly(new FirstN(3, "PV", true));
		// .aggregate(new Fields("count"), new Sum(), new Fields("sum"));

		return topology.build();

	}

	public static void main(String[] args) throws Exception {
		Config conf = new Config();
		// Config.TOPOLOGY_MAX_SPOUT_PENDING: 这个设置一个spout
		// task上面最多有多少个没有处理的tuple（没有ack/failed）回复，
		// 我们推荐你设置这个配置，以防止tuple队列爆掉。
		conf.setMaxSpoutPending(20);
		if (args.length == 0) {
			LocalDRPC drpc = new LocalDRPC();
			LocalCluster cluster = new LocalCluster();
			cluster.submitTopology("wordCounter", conf, buildTopology(drpc));
			for (int i = 0; i < 100; i++) {
				System.err.println("DRPC RESULT: "
						+ drpc.execute("GetPV",
								"2014-01-07 2014-01-08 2014-01-09 2014-01-10"));
				Thread.sleep(1000);
			}
		} else {
			conf.setNumWorkers(3);
			StormSubmitter.submitTopology(args[0], conf, buildTopology(null));
		}
	}
}
